package com.bjbn.service.impl;

import com.bjbn.service.KafkaAlarmPushService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Properties;

@Slf4j
@Service
public class KafkaAlarmPushServiceImpl implements KafkaAlarmPushService {

    @Value("${kafka.bootstrap-servers:10.120.78.214:32708}")
    private String bootstrapServers;

    @Value("${kafka.alarm-topic:BMS_ALARM_DATA}")
    private String alarmTopic;

    private KafkaProducer<String, String> producer;

    @PostConstruct
    public void init() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 可根据需要添加更多配置
        producer = new KafkaProducer<>(props);
        log.info("KafkaAlarmPushService初始化完成，服务器：{}，主题：{}", bootstrapServers, alarmTopic);
    }

    @Override
    public void pushAlarm(String alarmJson) {
        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(alarmTopic, alarmJson);
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    log.error("推送报警数据到Kafka失败", exception);
                } else {
                    log.info("报警数据已推送到Kafka，分区：{}，偏移量：{}", metadata.partition(), metadata.offset());
                }
            });
        } catch (Exception e) {
            log.error("推送报警数据到Kafka异常", e);
        }
    }
} 